package com.cl.cloud.binlog.factory;

import cn.hutool.core.util.IdUtil;
import com.cl.cloud.binlog.content.BinLogApplicationContext;
import com.cl.cloud.binlog.handler.PositionHandler;
import com.cl.cloud.binlog.listener.BinlogLifecycleListener;
import com.cl.cloud.binlog.listener.MultiEventHandlerListener;
import com.cl.cloud.binlog.model.BinlogPositionEntity;
import com.cl.cloud.binlog.parser.CommonEventParserDispatcher;
import com.cl.cloud.binlog.properties.DbConfig;
import com.cl.cloud.binlog.tableMeta.TableMetaFactory;
import com.cl.core.common.exception.ServerException;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @description: binlog监听客户端抽象工厂
 * @author: liuzijian
 * @time: 2022-07-08 11:09
 */
@Slf4j
public abstract class AbstractBinaryLogClientFactory implements BinaryLogClientFactory {


    private final ConcurrentHashMap<DbConfig, BinaryLogClient> cache = new ConcurrentHashMap<>();

    private final BinLogApplicationContext context;

    public AbstractBinaryLogClientFactory(BinLogApplicationContext context) {
        this.context = context;
    }

    @Override
    public BinaryLogClient getClient(DbConfig dbConfig) throws ServerException {
        BinaryLogClient client = this.getCachedClient(dbConfig);
        if (client != null) {
            return client;
        }
        //创建客户端
        client = new BinaryLogClient(
                dbConfig.getHost(),
                dbConfig.getPort(),
                dbConfig.getUserName(),
                dbConfig.getPassword()
        );
        /**
         * 获取binlog位点处理器
         */
        PositionHandler positionHandler = context.getPositionHandler();
        /**
         * 处理binlog位点信息
         * 从记录的position重新开始读取binlog
         */
        if (positionHandler != null) {
            BinlogPositionEntity positionEntity = positionHandler.getPosition(dbConfig);
            if (positionEntity != null && !StringUtils.isBlank(positionEntity.getBinlogName()) && positionEntity.getPosition() != null) {
                log.info("binlog位点{}", positionEntity.toString());
                client.setBinlogFilename(positionEntity.getBinlogName());
                long position = positionEntity.getPosition() != null ? positionEntity.getPosition() : 0L;
                client.setBinlogPosition(position);
            } else {
                log.info("binlog位点为空");
            }
        }
        /**
         * 处理数据库类型的数据
         */
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,//date和datetime类型转为long
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY//char和binary类型转为byte数组
        );
        client.setEventDeserializer(eventDeserializer);
        //设置slave的serverId，不同集群中，两个机器不能相同
        client.setServerId(IdUtil.getSnowflake().nextId());
        //初始化事件处理器
        MultiEventHandlerListener multiEventHandlerListener = new MultiEventHandlerListener(context, dbConfig);
        client.registerEventListener(multiEventHandlerListener);
        client.registerLifecycleListener(new BinlogLifecycleListener());
        cache.put(dbConfig, client);
        return client;
    }

    public BinaryLogClient getCachedClient(DbConfig dbConfig) {
        return cache.get(dbConfig);
    }

    @Override
    public void close() {
        for (BinaryLogClient client : cache.values()) {
            if (client.isConnected()) {
                try {
                    client.disconnect();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
